#include <cstdio>
#include <ctime>
#include <cstdlib>
#include <cstring>
#include <set>
#include <pthread.h>
#include "BTreeIndex.h"
#include "time_counter.h"
#include "LRUCache.h"
#include "Bitcask.h"
#include "Log.h"
#include "Objects.h"

bool running = true;
int record_size = 64;
int op;
SpinLock oplock;

int rand_num(int range)
{
    int a = rand() % 1000;
    a = a * 1000 + rand() % 1000;
    a = a * 1000 + rand() % 1000;
    a %= range;
    return a;
}


struct Task
{
    BtreeIndex *main_index;
    LRUCache *main_cache;
    Bitcask *bitcask;

    int *keys;
    Item **items;
    int start;
    int count;
};

void *thread_proc_set(void *arg)
{
    Task *task = reinterpret_cast<Task*>(arg);
    char *val = new char[128*1024];
    
    for(int i = task->start; i < task->start + task->count; ++i)
    {
        Item *cur_item = (Item*)mymalloc(sizeof(Item) + 12);

        u8 key_len = sprintf(cur_item->data, "key%09d", task->keys[i]);
        cur_item->Init(key_len, record_size);
        *(u64*)(&val[0]) = task->keys[i];
        if (record_size <= sizeof(void*))
        {
            cur_item->inner_data = 1;
            memcpy(&(cur_item->cache), val, record_size);
        }
        else
        {
            if (task->main_cache->SetItem(cur_item, val))
                printf("set cache failed!\n");
        }

        task->items[i] = cur_item;
        cur_item->real_pos = task->bitcask->Put(&cur_item->data[0], key_len, val, record_size);
        
        if (task->main_index->Set(cur_item) == NULL)
        {
            printf("set err %d\n", i);
        }
        cur_item->ReleaseRef();

        oplock.Lock();
        op++;
        oplock.UnLock();
    }

    delete []val;
    return NULL;
}


void *thread_proc_get(void *arg)
{
    Task *task = reinterpret_cast<Task*>(arg);
    Item *a = reinterpret_cast<Item*>(mymalloc(sizeof(Item)+31));
    Item* res;
    int hit = 0;
    char *val = new char[128*1024];

    for(int i = task->start; i < task->start + task->count; ++i)
    {
        u8 key_len = sprintf(&a->data[0], "key%09d", task->keys[i]);
        a->SetKeyLength(key_len);

        if ((res = task->main_index->Get(a)) == NULL)
        {
            printf("Not found err! %d  ", task->keys[i]);
            continue;
        }

        if (res->inner_data)
        {
            *(u64*)(&val[0]) = (u64)(res->cache);
            hit++;
        }
        else if (task->main_cache->GetItem(res, val) < 0)
        {
            if (task->bitcask->Get(val, res->ValLength(), res->real_pos) < 0)
            {
                printf("Get read error!");
            }   
        }
        else
        {
            hit++;
        }

        res->ReleaseRef();

        if (*(u64*)(&val[0]) != task->keys[i])
        {
            printf("Not match error! real: %llu  should %llu", *(u64*)(&val[0]), task->keys[i]);
        }

        oplock.Lock();
        op++;
        oplock.UnLock();
    }

    delete []val;

    myfree(a);
    printf("cache hit ratio: %.2f%%\n", (float)hit/task->count*100);
    return NULL;
}

void *thread_proc_del(void *arg)
{
    Task *task = reinterpret_cast<Task*>(arg);
    Item *a = reinterpret_cast<Item*>(mymalloc(sizeof(Item)+31));
    printf("Operation range: [%d,%d]\n", task->start, task->start + task->count - 1);

    for(int i = task->start; i < task->start + task->count; ++i)
    {
        u8 key_len = sprintf(&a->data[0], "key%09d", task->keys[i]);
        a->SetKeyLength(key_len);
        Item *item = task->main_index->Delete(a);

        if (item == NULL)
            printf("Not found err! %d  ", task->keys[i]);
        else
        {
            task->bitcask->Delete(item->data, item->KeyLength(), item->ValLength(), item->real_pos);
            if (!item->inner_data) task->main_cache->DeleteItem(item);
            item->Delete();
        }
        oplock.Lock();
        op++;
        oplock.UnLock();
    }
    myfree(a);
    return NULL;
}

void *thread_op_speed(void *arg)
{
    int last = 0;
    Bitcask *bitcask = reinterpret_cast<Bitcask*>(arg);

    while(1)
    {
        printf("ops: %d\n", op - last);
        last = op;
        sleep(1);
    }
    return NULL;
}

void *thread_proc_randomop(void *arg)
{
    Task *task = reinterpret_cast<Task*>(arg);
    Item *a = reinterpret_cast<Item*>(mymalloc(sizeof(Item)+30));

    for(int progress = task->start, i = 0; i < task->count * 3; ++i)
    {
        int op = task->keys[i/3] % 13;

        if (progress == task->start || op < 4 && progress < task->count + task->start)
        {
            if (task->main_index->Set(task->items[progress]) < 0)
            {
                printf("set err %d\n", i);
            }
            progress++;
        }
        else if (op < 10)
        {
            int idx = task->keys[progress] * task->keys[progress-1] % progress - task->start + task->start;//replace rand()with lock
            u8 key_len = sprintf(&a->data[0], "key%09d", task->keys[idx]);
            a->SetKeyLength(key_len);
            task->main_index->Get(a);

            /*if ( == a)
            {
                printf("Not found err!");
            }*/
        }
        else 
        {
            int idx = task->keys[progress] * task->keys[progress-1] % progress - task->start + task->start; //replace rand()with lock
            u8 key_len = sprintf(&a->data[0], "key%09d", task->keys[idx]);
            a->SetKeyLength(key_len);
            task->main_index->Delete(a);

            /*if (item == NULL)
                printf("Not found err!");
            else;*/
            //       memset(item, 0, sizeof(Item)+50);
        }
    }

    myfree(a);
    return NULL;
}


int main(int argc, char *argv[])
{
    BtreeIndex main_index;
    LRUCache main_cache;
    Bitcask bitcask;
    TimeCounter tc;
    void *tret;
    bool reopen = false;
    
    int key_num = 1000 * 1000;
    int randomize = 0;
    int thread_num = 1;

    printf("Usage: %s key_num randomize thread_num record_size\n", argv[0]);


#ifdef PALLOCATOR
    printf("Have palloc!\n");
    painit();
#endif

    if (argc > 1)
        key_num = atoi(argv[1]);

    if (argc > 2)
        randomize = atoi(argv[2]);

    if (argc > 3)
        thread_num = atoi(argv[3]);

    if (argc > 4)
        record_size = atoi(argv[4]);

    int *keys = new int[key_num];
    Item **items = new Item*[key_num];
    Task *tasks = new Task[thread_num];


    for(int i = 0; i < key_num; ++i)
        keys[i] = i;

    srand(time(NULL));
    if (randomize)
    for(int i = 0; i < key_num*2; ++i)
    {
        int a = rand_num(key_num), b = rand_num(key_num), c;
        c = keys[a];
        keys[a] = keys[b];
        keys[b] = c;
    }

    int last_start = 0;
    for(int i = 0; i < thread_num; ++i)
    {
        tasks[i].main_cache = &main_cache;
        tasks[i].bitcask = &bitcask;
        tasks[i].main_index = &main_index;
        tasks[i].keys = keys;
        tasks[i].items = items;
        tasks[i].start = last_start;
        tasks[i].count = (key_num - last_start)/(thread_num - i);
        last_start += tasks[i].count;
    }

    /*for(int i = 0; i < key_num; ++i)
    {
        Item *a = reinterpret_cast<Item*>(mymalloc(sizeof(Item)+30));
        u8 key_len = sprintf(&a->data[0], "key%09d", keys[i]);
        a->SetKeyLength(key_len);
        items[i] = a;
    }*/
    printf("make random time: %d  example: %d %d %d\n", tc.GetTimeMilliS(),
        keys[0], keys[1], keys[2]);

    tc.reset();

    main_index.CompareFunc = item_compare;
    main_index.Init();
    main_cache.Init(128*MB);
    LogInit(NULL, LINFO);
    Item::InitGlobalLock();
    
    bitcask.Init("./data/", &main_index); 
    if (main_index.GetElementNum() > 0)
        reopen = true;
    oplock.Init();
    printf("open db time time: %d  -- elementnum: %d\n", tc.GetTimeMilliS(), main_index.GetElementNum());

    if (main_index.GetElementNum() == 1)
    {
        printf("The only key is: %12s\n", reinterpret_cast<Item*>(main_index.leaf_head->ptrs[0])->data);
    }
    tc.reset();
    
    //==============================================

    pthread_t tids[thread_num];
    pthread_t tid_aux;

    //if (mixed_test)
    if (0)
    {
        for(int i = 1; i < thread_num; ++i)
        {
            pthread_create(&tids[i], NULL, thread_proc_randomop, &tasks[i]);
        }

        thread_proc_randomop(&tasks[0]);

        for(int i = 1; i < thread_num; ++i)
        {
            void *tret;
            pthread_join(tids[i], &tret);
        }

        printf("mixed time: %d\n", tc.GetTimeMilliS());
        tc.reset();
        return 0;
    }

    running = true;
    pthread_create(&tid_aux, NULL, thread_op_speed, &bitcask);
   

        for(int i = 1; i < thread_num ; ++i)
        {
            pthread_create(&tids[i], NULL, thread_proc_set, &tasks[i]);
        }

        thread_proc_set(&tasks[0]);

        for(int i = 1; i < thread_num; ++i)
        {
            void *tret;
            pthread_join(tids[i], &tret);
        }
    

    printf("set time: %d\n", tc.GetTimeMilliS());
    
    //running = false;
    //pthread_join(tid_aux, &tret);
    tc.reset();

    for(int i = 1; i < thread_num; ++i)
    {
        pthread_create(&tids[i], NULL, thread_proc_get, &tasks[i]);
    }

    thread_proc_get(&tasks[0]);

    for(int i = 1; i < thread_num; ++i)
    {
        void *tret;
        pthread_join(tids[i], &tret);
    }
    
    printf("get time: %d\n", tc.GetTimeMilliS());

    //running = false;
    //pthread_join(tid_aux, &tret);

    tc.reset();

    for(int i = 1; i < thread_num; ++i)
    {
        pthread_create(&tids[i], NULL, thread_proc_del, &tasks[i]);
    }

    thread_proc_del(&tasks[0]);

    for(int i = 1; i < thread_num; ++i)
    {
        void *tret;
        pthread_join(tids[i], &tret);
    }

    printf("delete time: %d\n", tc.GetTimeMilliS());
    
    sleep(11);
    tc.reset();
    bitcask.Close();
    printf("Close time: %d\n", tc.GetTimeMilliS());

    return 0;
}
